package com.project.platformcommon.kafka;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.project.platformcommon.Bean.EmailBean;
import com.project.platformcommon.constants.Constants;
import com.project.platformcommon.email.SendEmail;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.util.Optional;

@Service
@Slf4j
public class EmailConsumer {
    @Autowired
    private SendEmail sendEmail;
    private static Gson gson = new GsonBuilder().create();

    /**
     * 监听一个Kafka 主题
     **/
    @KafkaListener(topics = Constants.EMAIL_TOPIC)
    public void receiveMessageFromKafka(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        log.info("监听消息,MailUid:{}", gson.fromJson(String.valueOf(record.value()), EmailBean.class).getFrom());

        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            sendEmail.sendVerCodeMail(gson.fromJson(String.valueOf(record.value()), EmailBean.class));
        }
        ack.acknowledge();
    }
}
